package com.showdor.springboot.kafka.gjjl;

import com.alibaba.fastjson.JSON;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Async;

import java.util.Arrays;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 数据模拟
 */
@Slf4j
public class KafkaTestDataSender {
    private static String inputTopic = "local-monitoring-logs-topic";

    public static void main(String... args) {
        // Kafka配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.139.186:32092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        try {
            int i = 0;
            while (true) {
                if (i > Integer.MAX_VALUE - 1000) {
                    i = 0;
                }
                i++;
                int sbId = i % 3;
                GjzxJcsjjl jcsjjl = new GjzxJcsjjl();
                jcsjjl.setJcdx("1"); // 1-设备
                jcsjjl.setDxid("device_00" + sbId);
                jcsjjl.setDxmc("测试设备" + sbId);
                Date fixedDate = new Date();
                jcsjjl.setJcsj(fixedDate); // 使用固定时间确保在同一窗口
                jcsjjl.setJczb(Arrays.asList("temperature", "humidity").get(i % 2)); // 指标
                jcsjjl.setJczbz(java.lang.String.valueOf(25.0 + (i % 20) * 0.5)); // 25.0, 25.5, ..., 29.5
                jcsjjl.setZbzzt(Arrays.asList(0, 1).get(i % 20 == 0 ? 1 : 0)); // 1-正常
                jcsjjl.setDelFlag("0");

                // 转换为JSON字符串
                String jsonString = JSON.toJSONString(jcsjjl);

                // 创建消息记录
                ProducerRecord<String, String> record = new ProducerRecord<>(inputTopic, jsonString);

                // 发送消息
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            log.info("消息发送成功: Topic=" + recordMetadata.topic() + ", Partition=" + recordMetadata.partition() + ", Offset=" + recordMetadata.offset());
                        } else {
                            log.error("消息发送失败: " + e.getMessage());
                        }
                    }
                });

                // 延迟100毫秒发送下一条
                int delay = ThreadLocalRandom.current().nextInt(1000, 5000);
                Thread.sleep(delay);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}
